package com.zsp.coupon.service.imp;


import com.alibaba.fastjson.JSON;
import com.zsp.coupon.constant.Constant;
import com.zsp.coupon.constant.CouponStatus;
import com.zsp.coupon.dao.CouponDao;
import com.zsp.coupon.entity.Coupon;
import com.zsp.coupon.exception.CouponException;
import com.zsp.coupon.feign.SettlementClient;
import com.zsp.coupon.feign.TemplateClient;
import com.zsp.coupon.service.IRedisService;
import com.zsp.coupon.service.IUserService;
import com.zsp.coupon.vo.*;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.util.*;
import java.util.stream.Collectors;


/**
 * <h1>用户服务相关的接口实现</h1>
 * 所有的操作过程, 状态都保存在 Redis 中, 并通过 Kafka 把消息传递到 MySQL 中
 * 为什么使用 Kafka, 而不是直接使用 SpringBoot 中的异步处理 ?
 * Created by Qinyi.
 */

@Service
@Slf4j
public class UserviceImpl implements IUserService {

    @Autowired
    /** Coupon Dao */
    private final CouponDao couponDao;

    @Autowired
    /** Redis 服务 */
    private final IRedisService redisService;

    @Autowired
    /** 模板微服务客户端 */
    private final TemplateClient templateClient;

    @Autowired
    /** 结算微服务客户端 */
    private final SettlementClient settlementClient;

    @Autowired
    /** Kafka 客户端 */
    private final KafkaTemplate<String, String> kafkaTemplate;

    public UserviceImpl(CouponDao couponDao,
                        IRedisService redisService,
                        TemplateClient templateClient,
                        SettlementClient settlementClient,
                        KafkaTemplate<String, String> kafkaTemplate) {
        this.couponDao = couponDao;
        this.redisService = redisService;
        this.templateClient = templateClient;
        this.settlementClient = settlementClient;
        this.kafkaTemplate = kafkaTemplate;
    }


    /**
     * <h2>根据用户 id 和状态查询优惠券记录</h2>
     * @param userId 用户 id
     * @param status 优惠券状态
     * @return {@link Coupon}s
     */
    @Override
    public List<Coupon> findCouponByStatus(Long userId, Integer status)
            throws CouponException {
        List<Coupon> curCached =redisService.getCacheCoupons
                (userId,status);
        List<Coupon> pretarget;

        if (CollectionUtils.isNotEmpty(curCached)) {
            log.debug("coupon cache is not empty: {}, {}", userId, status);
            pretarget = curCached;
        } else  {
            log.debug("coupon cache is empty, get coupon from db: {}, {}",
                userId,status);

            List<Coupon> dbCoupons = couponDao.findAllByUserIdAndStatus(
                userId, CouponStatus.of(status)
            );

        // 如果数据库中没有记录, 直接返回就可以, Cache 中已经加入了一张无效的优惠券
        if (CollectionUtils.isEmpty(dbCoupons)) {
            log.debug("current user do not have coupon: {}, {}", userId, status);
                return dbCoupons;
        }

            // 填充 dbCoupons的 templateSDK 字段
            Map<Integer,CouponTemplateSDK> id2TemplateSDK =
                        templateClient.findIds2TemplateSDK(
                                dbCoupons.stream()
                                        .map(Coupon::getTemplateId)
                                        .collect(Collectors.toList())
                        ).getData();
                    dbCoupons.forEach(
                            dc ->dc.setTemplateSDK(
                            id2TemplateSDK.get(dc.getTemplateId())
                            )
                    );

            // 数据库中存在记录

            pretarget =dbCoupons;
            // 将记录写入 Cache
                redisService.addCouponToCache(userId,pretarget,status);
        }

        // 将无效优惠券剔除
            pretarget = pretarget.stream()
                    .filter(c ->c.getId() != -1)
                    .collect(Collectors.toList());

        // 如果当前获取的是可用优惠券, 还需要做对已过期优惠券的延迟处理
                if (CouponStatus.of(status) == CouponStatus.USEABLE) {
                    CouponClassify classify = CouponClassify.classify(pretarget);
                    // 如果已过期状态不为空, 需要做延迟处理
                    if (CollectionUtils.isNotEmpty(classify.getExpired())) {
                        log.info("Add Expired Coupons To Cache From FindCouponsByStatus: " +
                                "{}, {}", userId, status);
                            redisService.addCouponToCache(
                                    userId,classify.getExpired(),
                                    CouponStatus.EXPIRED.getCode()
                            );

                        // 发送到 kafka 中做异步处理
                            kafkaTemplate.send(Constant.TOPIC,
                                    JSON.toJSONString( new CouponKafakMessage(
                                            CouponStatus.EXPIRED.getCode(),
                                            classify.getExpired().stream()
                                                    .map(Coupon::getId)
                                                    .collect(Collectors.toList())
                                    ))
                            );
                    }
                    return classify.getUable();
        }
        return pretarget;
    }


    /**
     * <h2>根据用户 id 查找当前可以领取的优惠券模板</h2>
     * @param userId 用户 id
     * @return {@link CouponTemplateSDK}s
     */
    @Override
    public List<CouponTemplateSDK> findAllAviaiableTemplate(Long userId)
            throws CouponException {
        long curTime = new Date().getTime();
        List<CouponTemplateSDK> templateSDKS =
                templateClient.findAllUsableTemplate().getData();

        log.debug("Find All Template(From TemplateClient) Count: {}",
                templateSDKS.size());

        // 过滤过期的优惠券模板
            templateSDKS = templateSDKS.stream().filter(
                    t ->t.getRule().getException().getDeadline() >curTime
            ).collect(Collectors.toList());

        log.info("Find Usable Template Count: {}", templateSDKS.size());

        // key 是 TemplateId
        // value 中的 left 是 Template limitation, right 是优惠券模板
            Map<Integer, Pair<Integer,CouponTemplateSDK>> limit2Template =
                    new HashMap<>(templateSDKS.size());
            templateSDKS.forEach(
                    t -> limit2Template.put(
                            t.getId(),
                           Pair.of(t.getRule().getLimitaion(),t)
                    )
            );

        List<CouponTemplateSDK> result =
                new ArrayList<>(limit2Template.size());

        List<Coupon> userUsableCoupons = findCouponByStatus(
                userId, CouponStatus.USEABLE.getCode()
        );

        log.debug("Current User Has Usable Coupons: {}, {}", userId,
                userUsableCoupons.size());

        // key 是 TemplateId
        Map<Integer, List<Coupon>> templateId2Coupons = userUsableCoupons
                .stream()
                .collect(Collectors.groupingBy(Coupon::getTemplateId));
        // 根据 Template 的 Rule 判断是否可以领取优惠券模板
        limit2Template.forEach((k, v) -> {

            int limitation = v.getLeft();
            CouponTemplateSDK templateSDK = v.getRight();

            if (templateId2Coupons.containsKey(k)
                    && templateId2Coupons.get(k).size() >= limitation) {
                return;
            }

            result.add(templateSDK);

        });

        return result;
    }


    /**
     * <h2>用户领取优惠券</h2>
     * 1. 从 TemplateClient 拿到对应的优惠券, 并检查是否过期
     * 2. 根据 limitation 判断用户是否可以领取
     * 3. save to db
     * 4. 填充 CouponTemplateSDK
     * 5. save to cache
     * @param request {@link AcquireTemplateRequest}
     * @return {@link Coupon}
     */
    @Override
    public Coupon acquireTemplate(AcquireTemplateRequest request) throws CouponException {
                Map<Integer,CouponTemplateSDK> id2Tempalte =
                        templateClient.findIds2TemplateSDK(
                                Collections.singletonList(
                                request.getTemplateSDK().getId()
                        )).getData();

                //优惠卷模板是需要存在的
            if (id2Tempalte.size() <= 0 ) {
                log.error("Can Not Acquire Template From TemplateClient: {}",
                        request.getTemplateSDK().getId());
                throw new CouponException("Can Not Acquire Template From TemplateClient");
                // 用户是否可以领取这张优惠券
            }
                List<Coupon> userUsableCoupons = findCouponByStatus(
                        request.getUserId(), CouponStatus.USEABLE.getCode());

                Map<Integer, List<Coupon>> templateId2Coupons = userUsableCoupons
                        .stream()
                        .collect(Collectors.groupingBy(Coupon::getTemplateId));

                if (templateId2Coupons.containsKey(request.getTemplateSDK().getId())
                        && templateId2Coupons.get(request.getTemplateSDK().getId()).size() >=
                        request.getTemplateSDK().getRule().getLimitaion()) {
                    log.error("Exceed Template Assign Limitation: {}",
                            request.getTemplateSDK().getId());
                    throw new CouponException("Exceed Template Assign Limitation");
                }
        // 尝试去获取优惠券码
        String couponCodes = redisService.tryToAcquireCouponCodeFromCache(
                request.getTemplateSDK().getId()
        );
                if (StringUtils.isEmpty(couponCodes)) {
                    log.error("Can Not Acquire Coupon Code: {}",
                request.getTemplateSDK().getId());
                    throw new CouponException("Can Not Acquire Coupon Code");
                }
                    Coupon newCoupon = new Coupon(
                            request.getTemplateSDK().getId(),request.getUserId(),
                        couponCodes,CouponStatus.USEABLE
                    );
                        newCoupon =couponDao.save(newCoupon);

        // 填充 Coupon 对象的 CouponTemplateSDK, 一定要在放入缓存之前去填充
            newCoupon.setTemplateSDK(request.getTemplateSDK());

        // 放入缓存中
        redisService.addCouponToCache(
                request.getUserId(),Collections.singletonList(newCoupon),
            CouponStatus.USEABLE.getCode()
        );
        return newCoupon;
    }


    /**
     * <h2>结算(核销)优惠券</h2>
     * 这里需要注意, 规则相关处理需要由 Settlement 系统去做, 当前系统仅仅做
     * 业务处理过程(校验过程)
     * @param info {@link SettlementInfo}
     * @return {@link SettlementInfo}
     */
    @Override
    public SettlementInfo settlement(SettlementInfo info ) throws CouponException {
        // 当没有传递优惠券时, 直接返回商品总价
        return null;
    }

    /**
     * <h2>保留两位小数</h2>
     * */

    private double retain2Decimals(double value) {

        // BigDecimal.ROUND_HALF_UP 代表四舍五入
        return new BigDecimal(value)
                .setScale(2, BigDecimal.ROUND_HALF_UP)
                .doubleValue();
    }
}
